package com.atguigu.flink.chapter07.timer;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/14 9:49
 */
public class Flink01_Timer_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        
        env.setParallelism(2);
        // 每隔5s计算这5s内的wordCount
        env
            .socketTextStream("hadoop162", 9999)
            .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
                @Override
                public void flatMap(String value,
                                    Collector<Tuple2<String, Long>> out) throws Exception {
                    for (String word : value.split(" ")) {
                        out.collect(Tuple2.of(word, 1L));
                    }
                }
            })
            .keyBy(f -> f.f0)
            .process(new KeyedProcessFunction<String, Tuple2<String, Long>, String>() {
                private long now;
                int count = 0;
                // 当定时器触发的时候, 会自动回调这个方法
                @Override
                public void onTimer(long timestamp,  // 定时器的时间
                                    OnTimerContext ctx,
                                    Collector<String> out) throws Exception {
                    out.collect("预警: 定时器触发 " + timestamp);
                    count = 0;
                    
                }
                
                @Override
                public void processElement(Tuple2<String, Long> input,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    
                    String word = input.f0;
                    if ("a".equals(word) && count == 0) {
                        
                        // 参数: 定时器的触发时间, 当处理时间大于等于这个时间的时候, 会触发定时器
                        now = System.currentTimeMillis();
                        System.out.println("将在这个时间触发定时器: " + (now + 5000));
                        
                        ctx.timerService().registerProcessingTimeTimer(now + 5000);
                        count++;
                    }else if(count == 1){
                        System.out.println(now + 5000 + " 定时器被取消");
                        ctx.timerService().deleteProcessingTimeTimer(now + 5000);
                    }
                    
                    
                    
                }
            })
            .print();
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
